package com.lee.trigger;

import com.lee.entity.SessionResult;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;

import java.util.HashMap;

public class SessionTimeWindowTrigger<W extends Window> extends Trigger<SessionResult, W> {

    HashMap<String, Integer> timeDstMap;
    HashMap<String, Integer> timeServiceMap;

    public SessionTimeWindowTrigger(HashMap<String, Integer> timeDstMap, HashMap<String, Integer> timeServiceMap) {
        this.timeDstMap = timeDstMap;
        this.timeServiceMap = timeServiceMap;
    }

    @Override
    public TriggerResult onElement(SessionResult element, long timestamp, W window, TriggerContext ctx) throws Exception {
        otherAddOne(element);
//        App.logger.info(map.keySet().toString());
        return TriggerResult.CONTINUE;
    }

    public void otherAddOne(SessionResult element) {
        if (timeDstMap.containsKey(element.getDstIP().toString())) {
            timeDstMap.put(element.getDstIP().toString(), timeDstMap.get(element.getDstIP().toString()) + 1);
        } else {
            timeDstMap.put(element.getDstIP().toString(), 1);
        }

        String protocol = element.getProtocol();
        if (timeServiceMap.containsKey(protocol)) {
            timeServiceMap.put(protocol, timeServiceMap.get(protocol) + 1);
        } else {
            timeServiceMap.put(protocol, 1);
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {

        return TriggerResult.FIRE_AND_PURGE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(W window, TriggerContext ctx) throws Exception {
        ctx.deleteProcessingTimeTimer(window.maxTimestamp());
    }
}
